#include "lsv_rcache_v2.h"
#include "lsv_log.h"

/*init read cache*/
int lsv_rcache_init_v2(lsv_volume_proto_t * volume_proto){
        lsv_s32_t err = 0;
        lsv_s32_t i = 0;
        lsv_rcache_t * rcache = NULL;

        volume_proto->rcache = (void *)malloc(sizeof(lsv_rcache_t));
        if(NULL == volume_proto->rcache){
                /*print log*/
                err = -ENOMEM;
                return err;
        }
        memset(volume_proto->rcache, 0, sizeof(lsv_rcache_t));

        rcache = (lsv_rcache_t *)volume_proto->rcache;
        rcache->head = 0;
        rcache->tail = 0;
        rcache->chunk_num = 0;
        rcache->total_chunk = LSV_RCACHE_CHUNK_NUM;
        rcache->chunk_id = (lsv_u64_t *) malloc(LSV_RCACHE_CHUNK_NUM * sizeof(lsv_u64_t));
        if(NULL == rcache->chunk_id){
                /*print log*/
                err = -ENOMEM;
                return err;
        }
        memset(rcache->chunk_id, 0, sizeof(lsv_u64_t) * LSV_RCACHE_CHUNK_NUM);
        rcache->chunk_buf = (lsv_s8_t **)malloc(sizeof(lsv_s8_t *) * LSV_RCACHE_CHUNK_NUM);
        if(NULL == rcache->chunk_buf){
                /*print log*/
                err = -ENOMEM;
                return err;
        }
        memset(rcache->chunk_buf, 0, sizeof(lsv_s8_t *) * LSV_RCACHE_CHUNK_NUM);

        for(i = 0; i < LSV_RCACHE_CHUNK_NUM; i++){
                rcache->chunk_buf[i] = (lsv_s8_t *) malloc(sizeof(lsv_s8_t) * LSV_CHUNK_SIZE);
                if(NULL == rcache->chunk_buf[i]){
                        /*print log*/
                        err = -ENOMEM;
                        return err;
                }
                memset(rcache->chunk_buf[i], 0, sizeof(lsv_s8_t) * LSV_CHUNK_SIZE);
        }
        lsv_rwlock_init(&rcache->rwlock);
        return err;
}
#if 0
/**/
int lsv_rcache_insert(lsv_volume_proto_t *volume_proto, lsv_u32_t chunk_id, lsv_s8_t *buf){
        lsv_s32_t err = 0;
        lsv_s32_t tail_idx = 0;
        lsv_s32_t chunk_num = 0;
        lsv_rcache_t *rcache = NULL;

        rcache = (lsv_rcache_t *)volume_proto->rcache;
        tail_idx = rcache -> tail;
        chunk_num = rcache -> chunk_num;

        memcpy(rcache->chunk_buf[tail_idx], buf, LSV_CHUNK_SIZE);
        rcache->chunk_id[tail_idx] = chunk_id;
        rcache->tail = (++tail_idx + LSV_RCACHE_CHUNK_NUM) % LSV_RCACHE_CHUNK_NUM;
        if(chunk_num < LSV_RCACHE_CHUNK_NUM)
                rcache->chunk_num ++;
        return err;
}
#endif

static int __check_crc(void *chunk_buf, uint64_t lba, uint32_t chunk_id, int page_idx) {
        YASSERT(page_idx >= 0 && page_idx < LSV_WBUF_PAGE_NUM - 1);

        lsv_log_proto_t log_proto;
        log_proto.log = chunk_buf;
        lsv_log_proto_link(&log_proto);

        uint32_t crc1 = log_proto.hlog[page_idx].crc;
        uint32_t crc2 = page_crc32(log_proto.slog[page_idx].page);
        YASSERT(crc1 == crc2);
        DINFO("lba %llu crc %u chunk_id %u page_idx %u\n", (LLU)lba, crc1, chunk_id, page_idx);

        return 0;
}

int lsv_rcache_page_search(lsv_volume_proto_t *volume_proto, uint64_t lba, lsv_u32_t chunk_id,
                lsv_s32_t chunk_offset,lsv_s32_t page_offset, lsv_s32_t size, buffer_t *append_buf){
        lsv_s32_t err = 0;
        lsv_s32_t i = 0;
        lsv_s32_t tail_idx = 0;
        lsv_s32_t chunk_num = 0;
        lsv_s32_t idx_start = 0;
        lsv_s32_t idx_end = 0;
        lsv_rcache_t *rcache = NULL;

        rcache = (lsv_rcache_t *)volume_proto->rcache;
        tail_idx = rcache -> tail;
        chunk_num = rcache -> chunk_num;

        if(chunk_num < LSV_RCACHE_CHUNK_NUM){
                idx_end = tail_idx;
        }else{
                idx_end = LSV_RCACHE_CHUNK_NUM;
        }

        YASSERT(lba % LSV_PAGE_SIZE == 0);
        YASSERT(chunk_offset % LSV_PAGE_SIZE == 0);

        for(i = idx_start; i < idx_end; i++){
                if (chunk_id == rcache->chunk_id[i]){
#if WBUF_CHECK_CRC
                        __check_crc((void *)rcache->chunk_buf[i], lba, chunk_id, chunk_offset / LSV_PAGE_SIZE - 1);
#endif
                        mbuffer_appendmem(append_buf, rcache->chunk_buf[i] + chunk_offset+page_offset, size);
                        return size;
                }
        }
 
	return 0;
}

int lsv_rcache_load_chunk(lsv_volume_proto_t *volume_proto, uint64_t lba, lsv_u32_t chunk_id,
                lsv_s32_t chunk_offset,lsv_s32_t page_offset, lsv_s32_t size, buffer_t *append_buf){
        lsv_s32_t err = 0;
        lsv_s32_t i = 0;
        lsv_s32_t tail_idx = 0;
        lsv_s32_t chunk_num = 0;
        lsv_s32_t idx_start = 0;
        lsv_s32_t idx_end = 0;
        lsv_rcache_t *rcache = NULL;

	rcache = (lsv_rcache_t *)volume_proto->rcache;
        tail_idx = rcache -> tail;
        chunk_num = rcache -> chunk_num;

	rcache->chunk_id[tail_idx] = 0;
	err = lsv_log_read(volume_proto, chunk_id, rcache->chunk_buf[tail_idx]);
        if(err){
                DERROR("call lsv_log_slog_read read chunk_id[%d] error:%d\n", tail_idx, err);
                return err;
        }

#if WBUF_CHECK_CRC
        __check_crc((void *)rcache->chunk_buf[tail_idx], lba, chunk_id, chunk_offset / LSV_PAGE_SIZE - 1);
#endif
	mbuffer_appendmem(append_buf,
                        rcache->chunk_buf[tail_idx] +  chunk_offset+page_offset, size);

	rcache->chunk_id[tail_idx] = chunk_id;
        if(chunk_num < LSV_RCACHE_CHUNK_NUM){
                rcache->chunk_num ++;
        }
	YASSERT(rcache->chunk_num <= LSV_RCACHE_CHUNK_NUM);
        rcache->tail = ++tail_idx  % LSV_RCACHE_CHUNK_NUM;

	return size;
}

/**
 *
 * @param volume_proto
 * @param lba
 * @param chunk_id
 * @param chunk_offset
 * @param page_offset
 * @param size
 * @param append_buf
 * @return
 */
int lsv_rcache_lookup_v2(lsv_volume_proto_t *volume_proto, uint64_t lba, lsv_u32_t chunk_id,
                lsv_s32_t chunk_offset,lsv_s32_t page_offset, lsv_s32_t size, buffer_t *append_buf){
        int err = 0;
        lsv_rcache_t *rcache = NULL;

        DINFO("lba %llu chunk_id %u chunk_offset %u page_offset %u size %u\n",
              (LLU)lba, chunk_id, chunk_offset, page_offset, size);

        if(LSV_CHUNK_NULL == chunk_id){
                DERROR("lookup rcache , chunk_id is 0, the parameter is wrong!\n");
                err = -EINVAL;
                return err;
        }

        rcache = (lsv_rcache_t *)volume_proto->rcache;

	err = lsv_rcache_page_search(volume_proto, lba, chunk_id, chunk_offset, page_offset, size, append_buf);
	if(err == size){
		return size;
	}

        DINFO("miss lba %llu loading chunk_id %u...\n", (LLU)lba, chunk_id);
        volume_proto->read_rc_miss++;

	lsv_wrlock(&rcache->rwlock);

	err = lsv_rcache_page_search(volume_proto, lba, chunk_id, chunk_offset, page_offset, size, append_buf);
	if(err == size){
                lsv_rwunlock(&rcache->rwlock);
		return size;
	}

	err = lsv_rcache_load_chunk(volume_proto, lba, chunk_id, chunk_offset, page_offset, size, append_buf);
	if(err != size){
		lsv_rwunlock(&rcache->rwlock);
		return err;
	}

	lsv_rwunlock(&rcache->rwlock);
        return size;
}

/*chunk_id == LSV_CHUNK_NULL, clean all chunks*/
int lsv_rcache_clean_v2(lsv_volume_proto_t * volume_proto, lsv_u32_t chunk_id){
        lsv_s32_t err = 0;
        lsv_s32_t i = 0;
        lsv_rcache_t * rcache = NULL;

        DINFO("chunk_id %u\n", chunk_id);

        rcache = (lsv_rcache_t *)volume_proto->rcache;
        for(i = 0; i < rcache->chunk_num; i++){
                if(rcache->chunk_id[i] == chunk_id || LSV_CHUNK_NULL == chunk_id){
        		lsv_wrlock(&rcache->rwlock);
                        rcache->chunk_id[i] = 0;
                        memset(rcache->chunk_buf[i], 0, sizeof(lsv_s8_t) * LSV_CHUNK_SIZE);
        		lsv_rwunlock(&rcache->rwlock);
                }
        }
        return err;
}

/**/
int lsv_rcache_release_v2(lsv_volume_proto_t * volume_proto){
        lsv_s32_t err = 0;
        lsv_s32_t i = 0;
        lsv_rcache_t * rcache = NULL;

        rcache = (lsv_rcache_t *)volume_proto->rcache;

        for(i = 0; i < LSV_RCACHE_CHUNK_NUM; i ++){
                free(rcache->chunk_buf[i]);
                rcache->chunk_buf[i] = NULL;
        	lsv_rwlock_destroy(&rcache->rwlock);
        }
        free(rcache->chunk_buf);
        rcache->chunk_buf = NULL;
        free(volume_proto->rcache);
        volume_proto->rcache = NULL;
        return err;
}
